package com.demo.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import com.demo.rocketmq.config.RocketCallbackConfig;
import com.demo.rocketmq.constants.RocketMqConstants;
import com.demo.rocketmq.entity.RetrySendMsg;
import com.demo.rocketmq.mapper.RetrySendMsgMapper;
import com.demo.rocketmq.utils.RocketMqUtils;
import jakarta.annotation.Resource;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
public class SendAndCallbackProducer {

    @Resource
    private RocketMQTemplate rocketMqTemplate;

    @Resource
    private RetrySendMsgMapper retrySendMsgMapper;

    /**
     * 发送带有回调标识的消息，消费端消费完之后把回调标识发给生产者，然后生产者根据回调标识更新数据
     * 生产者更新数据的类：{@link RocketCallbackConfig}
     * @param topic 主题
     * @param payload 消息体
     * @param callbackKey 回调key
     * @param callbackTopicTag 回调主题tag
     * @return 发生的消息结果
     */
    @Transactional(rollbackFor = Exception.class)
    public SendResult sendAndCallback(String topic, Object payload, @NonNull String callbackKey,
                                      String callbackTopicTag) {
        log.info("主题:{}，消息：{}，回调key：{}，回调tag：{}", topic, JSON.toJSONString(payload), callbackKey, callbackTopicTag);

        org.apache.rocketmq.common.message.Message msg = convertToRocketMessage(topic, payload, callbackKey);
        msg.getProperties().put(RocketMqConstants.CALLBACK_KEY, callbackKey);
        msg.getProperties().put(RocketMqConstants.CALLBACK_TOPIC_TAG, callbackTopicTag);

        try {
            // 保存发送消息记录
            save(topic, payload, callbackKey, callbackTopicTag);
            return send(msg);
        } catch (Exception e) {
            log.error("发送消息失败", e);
            throw new RuntimeException(e);
        }
    }

    private org.apache.rocketmq.common.message.Message convertToRocketMessage(String topic, Object payload, String hashKey) {

        Pair<String, String> pair = RocketMqUtils.splitTopic(topic);
        org.apache.rocketmq.common.message.Message msg =
                new org.apache.rocketmq.common.message.Message(pair.getLeft(), pair.getRight(), JSON.toJSONBytes(payload));
        if (StringUtils.isNotBlank(hashKey)) {
            msg.setKeys(hashKey);
        }
        return msg;
    }

    private void save(String topic, Object payload, @NonNull String callbackKey,
                      String callbackTopicTag) {
        RetrySendMsg retrySendMsg = new RetrySendMsg();
        retrySendMsg.setCallbackKey(callbackKey);
        retrySendMsg.setCallbackTopicTag(callbackTopicTag);
        retrySendMsg.setTopic(topic);
        if (payload instanceof String) {
            retrySendMsg.setMsg((String) payload);
        } else {
            retrySendMsg.setMsg(JSON.toJSONString(payload));
        }
        retrySendMsgMapper.insert(retrySendMsg);
    }

    private SendResult send(org.apache.rocketmq.common.message.Message msg) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        DefaultMQProducer producer = rocketMqTemplate.getProducer();
        SendResult sendResult = producer.send(msg);
        logSendResult(sendResult);
        return sendResult;
    }

    private void logSendResult(SendResult sendResult) {
        log.info("发送消息的msgId：{}， 发送状态：{}", sendResult.getMsgId(), sendResult.getSendStatus());
    }
}
